package com.deng.rtc.timer;

import java.util.List;
import java.util.TimerTask;
import java.util.TreeMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import com.deng.rtc.admin.AdminNodeService;
import com.deng.rtc.base.StreamLogParser;
import com.deng.rtc.util.ConfigProperty;
import com.deng.rtc.util.Constants;
import com.deng.rtc.util.FileUtil;

public class AdminNodeTimer extends TimerTask{
	// JDK8的文件读写工具类
	protected FileUtil fu = new FileUtil();
	// 接收来自CN的原始实时数据流队列
	protected static ConcurrentLinkedQueue<String> dataQu = new ConcurrentLinkedQueue<String>();
	// 计算任务延迟计算文件目录下文件所对应的文件名（时间戳）及数据个数
	private static TreeMap<Long,Integer> delayTaskIdDataNums = new TreeMap<Long,Integer>();
	// 优先计算的内存级延迟任务对应数据块队列（计算加工好的JSON格式）
	protected static ConcurrentLinkedQueue<List<String>> memDelayQu = new ConcurrentLinkedQueue<List<String>>();
	// 当前时刻的上批小批量计算任务状态(0空闲，1忙碌)
	private static int minBathJobStatus = 0;
	// 框架定义的日志解析接口(自行传入实现类)
	private static StreamLogParser slp;
	// AN中的任务分发器
	private AdminNodeService ans = new AdminNodeService();
	// 时间计数
	private long timerNum = 0;
	// 自定义日志解析类
	public static void setStreamLogParser(StreamLogParser streamLogParser){
		slp = streamLogParser;
	}
	// 接受CN所传过来的原始日志数据
	public static boolean addSteamData(List<String> uLogs){
		return dataQu.addAll(uLogs);
	}
	// 当前任务状态
	public static int getJobStatus(){
		return minBathJobStatus;
	}
	// 任务调度器核心实现模块
	@Override
	public void run() {
		// 获取当前原始日志数据大小
		int curLogNum = dataQu.size();
		System.out.println(ConfigProperty.getCurDateTime()+" curLogNum : "+curLogNum);
		// 只有大于0的时候才能执行计算
		if(curLogNum>0){
			// 用户自定义实现解析后的JSON格式数据列表
			List<String> userActionList = slp.parseLogs(dataQu, curLogNum);
			// 当前内存级延迟任务数量
			int memDelayTaskNum = memDelayQu.size();
			// 当前任务状态空闲时
			if(minBathJobStatus==0){
				// 内存里有延迟任务优先计算延迟任务
				if(memDelayTaskNum>0){
					// 为当前即将计算的任务分配延迟最久的数据集
					ans.setUserActions(memDelayQu.poll());
					// 启动后台线程执行小批量任务计算
					new Thread(new AdminConsole()).start();
					// 判定当前内存级延迟任务数量是否达到上限
					if(memDelayTaskNum<Constants.maxDelayTaskNum){
						// 将延迟最久的任务文件获取数据塞进内存及延迟任务数据集
						if(delayTaskIdDataNums.size()>0){
							while(memDelayQu.size()<Constants.maxDelayTaskNum && delayTaskIdDataNums.size()>0){
								// 延迟任务文件目录 文件名+延迟任务文件后缀名
								String taskFile = Constants.delayTaskDir + delayTaskIdDataNums.pollFirstEntry().getKey() + Constants.delayTaskFileSurfix;
								// 读取延迟任务文件对应数据
								List<String> rtJsonList = fu.readActions(taskFile);
								if(rtJsonList!=null){
									memDelayQu.add(rtJsonList);
								}
								// 删除已读取的延迟任务文件
								fu.delActionFile(taskFile);
							}
						}
						// 此时判定内存级延迟任务数是否达到上限
						if(memDelayQu.size()<Constants.maxDelayTaskNum){
							// 将当前最新加工的行为数据塞进内存级延迟任务集
							memDelayQu.add(userActionList);
						}else{
							long rtNum = ans.getCurrentTime();
							// 将当前最新加工行为数据写入延迟任务文件目录
							delayTaskIdDataNums.put(rtNum, curLogNum);
							fu.writeActions(userActionList, Constants.delayTaskDir+rtNum+Constants.delayTaskFileSurfix);
						}
					}else{
						// 当前内存及延迟任务已达到上限
						// 直接将最新数据集写入文件
						long rtNum = ans.getCurrentTime();
						delayTaskIdDataNums.put(rtNum, curLogNum);
						fu.writeActions(userActionList, Constants.delayTaskDir+rtNum+Constants.delayTaskFileSurfix);
					}
				}else{
					ans.setUserActions(userActionList);
					new Thread(new AdminConsole()).start();
				}
			}else{

				if(memDelayTaskNum<Constants.maxDelayTaskNum){
					// 塞入内存
					memDelayQu.add(userActionList);
				}else{
					// 写入文件
					long rtNum = ans.getCurrentTime();
					delayTaskIdDataNums.put(rtNum, curLogNum);
					fu.writeActions(userActionList, Constants.delayTaskDir+rtNum+Constants.delayTaskFileSurfix);
				}
				System.out.println(ConfigProperty.getCurDateTime()+" 上次计算任务还没有结束");
			}
		}

		// 每100次任务批量进行一次垃圾回收
		if(++timerNum%100==0){
			//todo 修改 gc 方法
			System.out.println("分布式计算任务周期频率更新 " +timerNum);
			System.gc();
		}
	}
	//
	public void reRun(){
		// 还有延迟任务
		if(memDelayQu.size()>0){
			System.out.println("AdminNodeTimer reRun method ......"+memDelayQu.size());
			// 获取内存延迟任务数集
			ans.setUserActions(memDelayQu.poll());
			// 开始计算延迟最久的数据集任务
			new Thread(new AdminConsole()).start();
			// 延迟任务文件存在时
			if(delayTaskIdDataNums.size()>0){
				String taskFile = Constants.delayTaskDir + delayTaskIdDataNums.pollFirstEntry().getKey() + Constants.delayTaskFileSurfix;
				List<String> rtJsonList = fu.readActions(taskFile);
				if(rtJsonList!=null){
					memDelayQu.add(rtJsonList);
				}
				// 删除延迟任务文件
				fu.delActionFile(taskFile); 
			}
		}
	}
	// 内部类实现小批量任务计算
	protected class AdminConsole implements Runnable{
		
		@Override
		public void run() {
			minBathJobStatus = 1;
			try{
				ans.run();
			} catch (Exception e) {
				System.err.println("分布式计算任务管理服务错误： "+e.getMessage());
			}
			minBathJobStatus = 0;
			// 判定内存延迟任务是否存在，若存在则一直不停的循环计算
			reRun();
		}
	}

}
